package cn.com.sparkle.paxos.net.callback;

import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Id;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.PrepareResponseBad;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.PrepareResponseGood;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Value;
//import cn.com.sparkle.paxos.model.Id;
import cn.com.sparkle.paxos.net.CallBack;
import cn.com.sparkle.paxos.net.NetNode;
import cn.com.sparkle.paxos.paxosinstance.PaxosInstance;
import cn.com.sparkle.paxos.util.IdComparator;

public class PrepareCallBack implements CallBack {
	private final static Logger logger = Logger
			.getLogger(PrepareCallBack.class);
	private int needResponseCount;
	private final int quorum;

	private int goodNum = 0;
	private int badNum = 0;

	private PaxosInstance paxosInstance;
	private volatile long maxPrepareId = -1;
	private volatile Id lastVotedId = Id.newBuilder().setAddress("")
			.setIncreaseId(-1).build();
	private volatile Value lastValue = null;
	private ReentrantLock lock = new ReentrantLock();

	public PrepareCallBack(int quorum, int needResponseCount,
			PaxosInstance paxosInstance) {
		this.needResponseCount = needResponseCount;
		this.quorum = quorum;
		this.paxosInstance = paxosInstance;
	}

	@Override
	public void call(NetNode nnode, MessagePackage value) {
		try {
			lock.lock();
			if (value.hasPrepareResponseBad()) {
				PrepareResponseBad response = value.getPrepareResponseBad();
				if (response.getLastPrepareId() > maxPrepareId) {
					maxPrepareId = response.getLastPrepareId();
				}
				++badNum;
			} else {
				PrepareResponseGood response = value.getPrepareResponseGood();
				if (IdComparator.getInstance().compare(lastVotedId,
						response.getLastVotedId()) == -1) {
					// if (lastVotedId.compareTo(response.lastVotedId) == -1) {
					lastVotedId = response.getLastVotedId();
					if (response.hasValue()) {
						lastValue = response.getValue();
					}
				}
				++goodNum;
			}
			--needResponseCount;
			checkCount();
		} finally {
			lock.unlock();
		}
	}

	private void checkCount() {
		logger.debug(paxosInstance.getInstanceId() +  "goodNum:" + goodNum + " badNum:" + badNum
				+ " needResponseCount:" + needResponseCount);
		if (goodNum == quorum) {
			++goodNum;// avoid invoking repeatedly
			paxosInstance.prepareSuccess(lastValue);
		} else if (goodNum < quorum && needResponseCount == 0) {
			if (badNum >= quorum) {
				paxosInstance.fail(maxPrepareId);
			} else {
				paxosInstance.fail(-1);
			}
		}
	}

	@Override
	public void fail(NetNode nnode) {
		try {
			lock.lock();
			--needResponseCount;
			checkCount();
		} finally {
			lock.unlock();
		}
	}

}
